package com.atguigu.chapter11.function;

import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.annotation.DataTypeHint;
import org.apache.flink.table.annotation.FunctionHint;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.table.functions.TableFunction;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * @Author lizhenchao@atguigu.cn
 * @Date 2021/6/21 9:32
 */
public class Flink01_Function_Table {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);
        
        DataStreamSource<String> stream = env.fromElements("hello world a", "atguigu hello b", "Hello");
        
        Table table = tEnv.fromDataStream(stream, $("line"));
        
        // 1. table API中使用
        // 1.1 使用内联方式
        /*table
//            .joinLateral(call(SplitUpperLen.class, $("line")))// 内连接
            .leftOuterJoinLateral(call(SplitUpperLen.class, $("line")))// 左连接
            .select($("line"), $("word"), $("len"))
            .execute()
            .print();*/
        
        // 1.2 注册之后使用
        // 略
        
        // 2. 在sql中使用
        tEnv.createTemporaryView("t1", table);
        
        tEnv.createTemporaryFunction("split_upper_len", SplitUpperLen.class);
        
        /*tEnv.sqlQuery("select " +
                          "* " +
                          "from t " +
                          "join lateral table(split_upper_len(line)) on true")
            .execute()
            .print();*/
       /* tEnv.sqlQuery("select " +
                          "* " +
                          "from t " +
                          "left join lateral table(split_upper_len(line)) on true")
            .execute()
            .print();*/
    
        // 内连接的第二种写法  select .. form a,b where a.id=b.id;  // 内连接
        tEnv.sqlQuery("select " +
                          "* " +
                          "from t1, lateral table(split_upper_len(line)) as t(w, le)")
            .execute()
            .print();
        
    }
    
    @FunctionHint(output = @DataTypeHint("ROW<word string, len int>"))
    public static class SplitUpperLen extends TableFunction<Row> {
        public void eval(String line) {
            
            String[] words = line.split(" ");  //hello word
            if (words.length < 2) {
                return;
            }
            for (String word : words) {
                // 这个方法是每调用一次, 就会得到一行
                collect(Row.of(word.toUpperCase(), word.length()));
            }
            
        }
        
    }
    
}

/*
"hello world"     hello   5
                  world   5
"atguigu hello"   atgugiu  7
                   hello   5
                
"Hello hello   ...


 */

